package com.sunwayworld.basemodule.common.eptstorage.acqconfig.net;

import com.sunwayworld.basemodule.common.eptstorage.acqconfig.bean.LimsEquiptAcqConfigInstBean;
import com.sunwayworld.basemodule.common.eptstorage.acqconfig.service.LimsEquiptAcqConfigInstService;
import com.sunwayworld.basemodule.common.eptstorage.acqconfig.service.LimsEquiptAcqConfigService;
import com.sunwayworld.basemodule.rm.equipment.bean.LimsEquipmentBean;
import com.sunwayworld.cloud.module.equipt.acqconfig.bean.CoreEquiptAcqConfigBean;
import com.sunwayworld.framework.exception.InvalidDataException;
import com.sunwayworld.framework.mybatis.mapper.MatchPattern;
import com.sunwayworld.framework.mybatis.mapper.SearchFilter;
import com.sunwayworld.framework.utils.ObjectUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

@Component
public class NetEptSenderHelper {
    private static final Logger logger = LogManager.getLogger(NetEptSenderHelper.class);
    //对于长期连不上的设备设置黑名单，重试3次连接不上，剔除
    public static final ConcurrentHashMap<String,NetEptSender> cacheMap = new ConcurrentHashMap<>(); //缓存设备连接
    public static final ConcurrentHashMap<String, Queue<LimsEquiptAcqConfigInstBean>> eptInstMap = new ConcurrentHashMap<>(); //缓存设备要发送的指令
    public static final ConcurrentHashMap<String, CoreEquiptAcqConfigBean> cacheEptAcqConfigMap = new ConcurrentHashMap<>(); //缓存设备配置

    public static final int maxInst = 200; //每个设备最大指令队列大小
    public static final ConcurrentHashMap<String,Integer> failMap = new ConcurrentHashMap<>(); //缓存失败设备

    @Autowired
    private LimsEquiptAcqConfigInstService limsEquiptAcqConfigInstService;
    @Autowired
    @Lazy
    private LimsEquiptAcqConfigService limsEquiptAcqConfigService;

    public static void failUp(CoreEquiptAcqConfigBean acqEptBean){
        String key = getKey(acqEptBean);
        Integer integer = failMap.get(key);
        if(integer>=Integer.MAX_VALUE){
            integer = 0;
        }
        failMap.put(key,integer==null?1:integer+1);
        logger.error(key+" 连接失败次数："+failMap.get(key));
    }

    public static void failClear(CoreEquiptAcqConfigBean acqEptBean){
        failMap.remove(getKey(acqEptBean));
    }

    public static final int failMax = 3;

    public static void putNetEptSender(CoreEquiptAcqConfigBean acqEptBean,NetEptSender eptSender){
        if(acqEptBean!=null && eptSender!=null && eptSender.isConnect()){
            String key = getKey(acqEptBean);
            cacheMap.put(key,eptSender);
        }
    }
    public NetEptSender getCacheNetEptSender(CoreEquiptAcqConfigBean acqEptBean){
        return cacheMap.get(getKey(acqEptBean));
    }

    public static void removeNetEptSender(CoreEquiptAcqConfigBean acqEptBean){
        if(acqEptBean!=null){
            String key = getKey(acqEptBean);
            cacheMap.remove(key);
        }
    }

    private  final ThreadPoolExecutor connectExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);

    public void testConnect(CoreEquiptAcqConfigBean acqEptBean){
        String key = acqEptBean.getIp()+":"+acqEptBean.getPort();
        NetEptSender netEptSender = cacheMap.get(key); //连接复用
        String netacqtype = acqEptBean.getExt$Item("NETACQTYPE");
        String acqCategory = acqEptBean.getAcqCategory();
        if(netEptSender==null || !netEptSender.isConnect()){
            logger.info("重新连接={},{}....",key,netacqtype);
            if(netEptSender!=null){
                netEptSender.close();
            }

            NetEptSender newNetEptSender = null;
            if ("net".equalsIgnoreCase(acqCategory)){
                if("tcp".equalsIgnoreCase(netacqtype)){
                    newNetEptSender = new NetEptTcpSender(acqEptBean);
                }else if("udp".equalsIgnoreCase(netacqtype)){
                    newNetEptSender = new NetEptUdpSender(acqEptBean);
                }
            }
            boolean connect = newNetEptSender.getConnection();
            if (!connect){
                logger.info("连接失败={},{}....",key,netacqtype);
            }
        }else{
            logger.info("复用连接={},{}....",key,netacqtype);
        }
    }

    @Transactional
    public synchronized String synSend(LimsEquipmentBean equiptBean, String instructionType){
        CoreEquiptAcqConfigBean acqConfig = limsEquiptAcqConfigService.selectDetail(equiptBean.getId().toString());
        List<LimsEquiptAcqConfigInstBean> tempEquiptAcqConfigInstBeans = limsEquiptAcqConfigInstService.selectListByFilter(SearchFilter.instance()
                .match("acqConfigId",acqConfig.getId()).filter(MatchPattern.EQ)
                .match("instructionType", instructionType).filter(MatchPattern.EQ));
        return synSend(acqConfig,tempEquiptAcqConfigInstBeans.get(0));
    }

    //同步发送数据
    @Transactional
    public synchronized String synSend(CoreEquiptAcqConfigBean acqEptBean, LimsEquiptAcqConfigInstBean instBean){
        if(acqEptBean==null || ObjectUtils.isEmpty(instBean)){
            throw new InvalidDataException("获取数据所需的配置信息为空");
        }
        String key = getKey(acqEptBean);
        testConnect(acqEptBean);//检查连接是否正常，若不正常则重新获取
        NetEptSender netEptSender = cacheMap.get(key);
        if(netEptSender==null){
            logger.error("netEptSender={} is null ",key);
            throw new InvalidDataException("连接失败！");
        }
        if(!netEptSender.isConnect()){
            logger.error("netEptSender={} is not connect ",key);
            throw new InvalidDataException("连接失败！");
        }
        String result = netEptSender.send(instBean);
        return result;
    }

    //异步发送指令，存储待发送的设备-指令队列 map
    @Transactional
    public synchronized void asynSend(CoreEquiptAcqConfigBean acqEptBean, List<LimsEquiptAcqConfigInstBean> instBeanList){
        if(acqEptBean==null || ObjectUtils.isEmpty(instBeanList)){
            return;
        }
        String key = getKey(acqEptBean);
        String netacqtype = acqEptBean.getExt$Item("NETACQTYPE");
        Queue<LimsEquiptAcqConfigInstBean> instBeanQueue = eptInstMap.get(key);
        if(instBeanQueue==null){
            instBeanQueue = new ConcurrentLinkedQueue<>();
        }
        for (LimsEquiptAcqConfigInstBean configInstBean : instBeanList) {
            if (!instBeanQueue.stream()
                    .filter(e -> !ObjectUtils.isEmpty(e.getId()))
                    .anyMatch(e -> e.getId().equals(configInstBean.getId()))) {//不存重复的指令
                instBeanQueue.offer(configInstBean);
            }
        }
        int size = instBeanQueue.size();
        logger.info("为设备={},{},缓存要发送的指令队列大小={}",key,netacqtype,size);
        if(size>=maxInst){
            for (int i = 0; i < instBeanList.size(); i++) {
                instBeanQueue.poll();
            }
            logger.info("设备={},{},缓存的指令数过大,超出{},移除旧的指令={}",key,netacqtype,maxInst,instBeanList.size());
        }
        eptInstMap.put(key,instBeanQueue);

/*        NetEptSender netEptSender = cacheMap.get(key); //连接复用
        logger.info("设备网口连接数={}....",cacheMap.keySet().size());
        if(netEptSender==null || !netEptSender.isConnect()){
            logger.info("重新连接={},{}....",key,netacqtype);
            if(netEptSender!=null){
                netEptSender.clearQueue();
                netEptSender.setConnect(false);
            }
            removeNetEptSender(acqEptBean);

            if("tcp".equalsIgnoreCase(netacqtype)){
                netEptSender = new NetEptTcpSender(acqEptBean);
            }else if("udp".equalsIgnoreCase(netacqtype)){
                netEptSender = new NetEptUdpSender(acqEptBean);
            }
            netEptSender.connect(); //异步连接，连接成功的回调中线程循环读发送的指令队列

        }else{
            logger.info("复用连接={},{}....",key,netacqtype);
        }*/
    }

    //由系统定时任务调度
    public void cycleCheckEptNetConnection(){
        synchronized (this){
            int corePoolSize = connectExecutor.getCorePoolSize();
            int poolSize = connectExecutor.getPoolSize();
            int activeCount = connectExecutor.getActiveCount();
            long queueSize = connectExecutor.getQueue().size();
            logger.info("定时检测设备net连接 connectExecutor,corePoolSize={},poolSize={},activeCount={},taskCount={}",corePoolSize,poolSize,activeCount,queueSize);
            Set<Map.Entry<String, CoreEquiptAcqConfigBean>> entries = cacheEptAcqConfigMap.entrySet();
            for (Map.Entry<String, CoreEquiptAcqConfigBean> entry : entries) {
                String key = entry.getKey();
                CoreEquiptAcqConfigBean acqEptBean = entry.getValue();
                if(acqEptBean!=null){
                    connectExecutor.submit(()->{
                        testConnect(acqEptBean);
                    });
                }
            }
        }
    }

    private static String getKey(CoreEquiptAcqConfigBean configBean){
        return configBean.getIp()+":"+configBean.getPort();
    }

}
